package netty.dubbo.server.server;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;
import netty.dubbo.server.registry.RegistryCenter;

import java.io.File;
import java.net.URL;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author: 程序员七哥
 * @date: 2021-12-19
 * @description: 服务启动类，用来发布服务
 */
public class RpcServer {

    // 容器类，用于存储服务名称和对应的实现类
    private Map<String, Object> registerMap = new HashMap<>();
    // 存储服务提供类名的列表
    private List<String> classCaches = new ArrayList<>();
    // 当前服务提供者所在的地址
    private String serviceAddress;

    /**
     * 服务发布程序，和 RPC 不同之处在于不仅在注册到本地的 Map 中，还要发布到 zk 注册中心
     * @param providerPackage
     * @throws Exception
     */
    public void publish(RegistryCenter registryCenter, String serviceAddress, String providerPackage) throws Exception {

        // 获取所有服务提供者的类名
        getProviderClass(providerPackage);
        // 注册到容器中
        doRegister(registryCenter,serviceAddress);

        this.serviceAddress = serviceAddress;


    }

    /**获取服务提供者的类名*/
    private void getProviderClass(String providerPackage) {
        // 通过包名获取文件资源路径
        final URL resource = this.getClass().getClassLoader().getResource(providerPackage.trim().replaceAll("\\.", "/"));
        // 通过路径获取目录
        File dir = new File(resource.getFile());
        for (File file: dir.listFiles()) {
            // 如果是目录则递归调用查找
           if (file.isDirectory()) {
               getProviderClass(providerPackage + "." + file.getName());
           } else if (file.getName().endsWith(".class")){
               final String fileName = file.getName().replace(".class", "").trim();
               classCaches.add(providerPackage + "." + fileName);
           }
        }
    }

    /**
     * 服务提供者进行注册
     */
    private void doRegister(RegistryCenter registryCenter, String serviceAddress) throws Exception {
        if (classCaches.size() == 0) {
            return;
        }
        for (String className : classCaches) {
            // 需要传入类的完全限定名，eg：Class.forName("java.lang.Thread")
            final Class<?> clazz = Class.forName(className);
            final String interfaceName = clazz.getInterfaces()[0].getName();
            registerMap.put(interfaceName,clazz.getDeclaredConstructor().newInstance());
            // 将服务接口和对应的提供者地址注册到 zk 注册中心
            registryCenter.register(interfaceName, serviceAddress);
        }
    }

    public void start() throws InterruptedException {
        NioEventLoopGroup parent = new NioEventLoopGroup();
        NioEventLoopGroup child = new NioEventLoopGroup();
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(parent,child)
                    // 指定用于存放请求的队列的长度，默认为50，其是Socket的标准参数
                    .option(ChannelOption.SO_BACKLOG,1024)
                    .childOption(ChannelOption.SO_KEEPALIVE,true)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            ChannelPipeline pipeline = socketChannel.pipeline();
                            pipeline.addLast(new ObjectEncoder());
                            pipeline.addLast(new ObjectDecoder(Integer.MAX_VALUE, ClassResolvers.cacheDisabled(null)));
                            pipeline.addLast(new RpcServerHandler(registerMap));
                        }
                    });
            final String[] strings = serviceAddress.split(":");
            String ip = strings[0];
            int port = Integer.parseInt(strings[1]);
            final ChannelFuture future = bootstrap.bind(ip,port).sync();
            System.out.println("服务提供者已启动：" + ip + port);
            future.channel().closeFuture().sync();
        } finally {
            parent.shutdownGracefully();
            child.shutdownGracefully();
        }
    }
}
